package com.example.kafka.service;

import com.example.kafka.config.KafkaConfig;
import com.example.kafka.model.Message;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;

@Slf4j
@Service
@RequiredArgsConstructor
public class KafkaProducerService {

    private final KafkaTemplate<String, String> kafkaTemplate;
    private final ObjectMapper objectMapper;

    /**
     * 发送消息
     *
     * @param content 消息内容
     * @return 消息对象
     */
    public Message sendMessage(String content) {
        try {
            // 构建消息对象
            Message message = Message.builder()
                    .id(UUID.randomUUID().toString())
                    .content(content)
                    .topic(KafkaConfig.TOPIC_TEST)
                    .sender("producer-demo")
                    .createTime(LocalDateTime.now())
                    .status("CREATED")
                    .build();

            // 转换为JSON字符串
            String messageJson = objectMapper.writeValueAsString(message);

            // 发送消息
            CompletableFuture<SendResult<String, String>> future = kafkaTemplate.send(
                    KafkaConfig.TOPIC_TEST,
                    message.getId(),
                    messageJson
            );

            // 异步处理发送结果
            future.whenComplete((result, ex) -> {
                if (ex == null) {
                    // 发送成功，更新消息状态和分区信息
                    message.setStatus("SENT");
                    message.setPartition(result.getRecordMetadata().partition());
                    message.setOffset(result.getRecordMetadata().offset());
                    log.info("消息发送成功: {}", message);
                } else {
                    // 发送失败
                    message.setStatus("FAILED");
                    log.error("消息发送失败: {}", message, ex);
                }
            });

            return message;
        } catch (Exception e) {
            log.error("消息发送异常", e);
            throw new RuntimeException("消息发送失败", e);
        }
    }

    /**
     * 发送简单消息
     */
    public void sendMessage(String topic, String message) {
        CompletableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, message);
        
        future.whenComplete((result, ex) -> {
            if (ex == null) {
                log.info("消息发送成功: topic = {}, message = {}, partition = {}", 
                    topic, message, result.getRecordMetadata().partition());
            } else {
                log.error("消息发送失败: topic = {}, message = {}", topic, message, ex);
            }
        });
    }

    /**
     * 发送带key的消息（相同key的消息会被发送到相同分区）
     */
    public void sendMessageWithKey(String topic, String key, String message) {
        CompletableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, key, message);
        
        future.whenComplete((result, ex) -> {
            if (ex == null) {
                log.info("带key的消息发送成功: topic = {}, key = {}, message = {}, partition = {}", 
                    topic, key, message, result.getRecordMetadata().partition());
            } else {
                log.error("带key的消息发送失败: topic = {}, key = {}, message = {}", 
                    topic, key, message, ex);
            }
        });
    }

    /**
     * 发送消息对象
     */
    public void sendMessageObject(String topic, String content) {
        try {
            Message message = Message.builder()
                    .id(UUID.randomUUID().toString())
                    .content(content)
                    .topic(topic)
                    .sender("system")
                    .createTime(LocalDateTime.now())
                    .status("SENT")
                    .build();

            String jsonMessage = objectMapper.writeValueAsString(message);
            CompletableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, message.getId(), jsonMessage);

            future.whenComplete((result, ex) -> {
                if (ex == null) {
                    message.setPartition(result.getRecordMetadata().partition());
                    message.setOffset(result.getRecordMetadata().offset());
                    message.setStatus("DELIVERED");
                    log.info("消息对象发送成功: {}", message);
                } else {
                    message.setStatus("FAILED");
                    log.error("消息对象发送失败: {}", message, ex);
                }
            });
        } catch (JsonProcessingException e) {
            log.error("消息序列化失败", e);
        }
    }
} 